package com.edu.learning.concurrent.sample.mode4;

import com.edu.learning.concurrent.FixedThreadPoolHelper;
import com.edu.learning.concurrent.ThreadPoolBuilder;
import com.google.common.util.concurrent.ThreadFactoryBuilder;

import java.util.Random;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

import static java.util.concurrent.TimeUnit.SECONDS;

/**
 * LinkedBlockingQueue、ArrayedBlockingQueue、SynchronousQueue、PriorityBlockingQueue和DelayQueue
 *
 * LinkedBlockingQueue和ArrayedBlockingQueue的共同点在于，
 * 它们都是一个FIFO（先入先出）队列，列头元素就是最老的元素，列尾元素就是最新的元素。
 * 元素总是从列尾插入队列，从列头取出队列。
 * 它们的不同之处在于，LinkedBlockingQueue的内部实现是一个链表，而ArrayedBlockingQueue的内部实现是一个原生数组。
 * 正如其他Java集合一样，链表形式的队列，其存取效率要比数组形式的队列高。
 * 但是在一些并发程序中，数组形式的队列由于具有一定的可预测性，因此可以在某些场景中获得更好的效率。
 * 另一个不同点在于，ArrayedBlockingQueue支持“公平”策略。
 * 若在构造函数中指定了“公平”策略为true，则阻塞在插入或取出方法的所有线程也会按照FIFO进行排队，这样可以有效避免一些线程被“饿死”。
 * BlockingQueue queue = new ArrayBlockingQueue<>(3, true);
 * 总体而言，LinkedBlockingQueue是阻塞队列的最经典实现，在不需要“公平”策略时，基本上使用它就够了。
 *
 * @author ll
 * @Date 2017/10/26 9:45
 */
public class BlockingQueueDemo {
    public static void main(String[] args) {
        BlockingQueue<String> blockingQueue = new LinkedBlockingQueue<>(3);

        //ExecutorService executorService = FixedThreadPoolHelper.getExecutorService("BlockingQueueDemo", false);

        ThreadFactory threadFactory =
                new ThreadFactoryBuilder().setNameFormat("BlockingQueueDemo-%d").build();
        ExecutorService executorService = new ThreadPoolExecutor(
                32,
                32,
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(1024),
                threadFactory,
                new ThreadPoolExecutor.CallerRunsPolicy());
        for (int i = 0; i < 5; i ++) {
            executorService.submit(new Producer("Producer " + i, blockingQueue));
        }

        for (int i = 0; i < 5; i ++) {
            executorService.submit(new Consumer("Consumer " + i, blockingQueue));
        }
        executorService.shutdown();
    }

    static class Producer implements Runnable {

        private final BlockingQueue<String> blockingQueue;
        private final String name;
        private static Random rand = new Random(47);
        private static AtomicInteger productID = new AtomicInteger(0);

        public Producer(String name, BlockingQueue<String> blockingQueue) {
            this.blockingQueue = blockingQueue;
            this.name = name;
        }

        @Override
        public void run() {
            try {
                StringBuilder sb = new StringBuilder();
                for (int i = 0; i < 10; i ++) {
                    SECONDS.sleep(rand.nextInt(5));
                    String str = "Product" + productID.getAndIncrement();
//                    sb.append("Product").append(productID.getAndIncrement()).append("\t");
                    blockingQueue.add(str);
                    //注意，这里得到的size()有可能是错误的
                    sb.append(name).append(" product ").append(str).append(", queue.size = ").append(blockingQueue.size());
                    System.out.println(sb);//name + " product " + sb + "\t , queue.size = " + blockingQueue.size()
                }
                System.out.println(name + " is over");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    static class Consumer implements Runnable {

        private final BlockingQueue<String> blockingQueue;
        private final String name;
        private static Random rand = new Random(47);

        public Consumer(String name, BlockingQueue<String> blockingQueue) {
            this.blockingQueue = blockingQueue;
            this.name = name;
        }

        @Override
        public void run() {
            try {
                StringBuilder sb = new StringBuilder();
                for (int i = 0; i < 10; i ++) {
                    SECONDS.sleep(rand.nextInt(5));
                    String str = blockingQueue.take();
                    //注意，这里得到的size()有可能是错误的
                    sb.append(name).append(" consume ").append(str).append(", queue size = ").append(blockingQueue.size());
                    System.out.println(sb);
                }
                System.out.println(name + " is over");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
